package com.corpgovernment.common.mq.config;

import com.corpgovernment.common.apollo.CommonApollo;
import com.ctrip.corp.obt.generic.utils.JsonUtils;
import com.corpgovernment.common.utils.LogSplicingUtils;
import com.ctrip.corp.obt.generic.utils.StringUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
@Slf4j
public class MQProducerConfiguration {

	@Autowired
	private CommonApollo commonApollo;

	@Bean
	public DefaultMQProducer getDefaultMQProducer() {
		StringBuilder logContext = new StringBuilder();
		try {
			LogSplicingUtils.addLogContext(logContext, "开始启动MQ生产者");

			String namesrvAddr = commonApollo.getProducerNamesrvAddr();
			LogSplicingUtils.addLogContext(logContext, "namesrvAddr：%s", namesrvAddr);
			if ("f".equals(namesrvAddr)) {
				return null;
			}

			String accessKey = commonApollo.getProducerAccessKey();
			String accessKeySecret = commonApollo.getProducerAccessKeySecret();
			LogSplicingUtils.addLogContext(logContext, "accessKey：%s, accessKeySecret：%s", accessKey, accessKeySecret);

			AclClientRPCHook rpcHook = null;
			if (StringUtils.isNotBlank(accessKey) && StringUtils.isNotBlank(accessKeySecret)) {
				rpcHook = new AclClientRPCHook(new SessionCredentials(accessKey, accessKeySecret));
			}

			DefaultMQProducer producer;
			if (rpcHook != null) {
				producer = new DefaultMQProducer(commonApollo.getProducerNamespace(),"CORP-GOVERNMENT", rpcHook, true, null);
				producer.setAccessChannel(AccessChannel.CLOUD);
				LogSplicingUtils.addLogContext(logContext, "namespace：%s", commonApollo.getProducerNamespace());
				LogSplicingUtils.addLogContext(logContext, "rpcHook not null");
			} else {
				producer = new DefaultMQProducer("CORP-GOVERNMENT");
				LogSplicingUtils.addLogContext(logContext, "rpcHook null");
			}

			producer.setNamesrvAddr(namesrvAddr);
			producer.setCreateTopicKey("AUTO_CREATE_TOPIC_KEY");
			//如果需要同一个jvm中不同的producer往不同的mq集群发送消息，需要设置不同的instanceName
			//producer.setInstanceName(instanceName);
			producer.setMaxMessageSize(commonApollo.getMaxMessageSize());
			producer.setSendMsgTimeout(commonApollo.getSendMsgTimeout());
			//如果发送消息失败，设置重试次数，默认为2次
			producer.setRetryTimesWhenSendFailed(commonApollo.getRetryTimesWhenSendFailed());
			producer.start();

			this.testPushMessage(producer, logContext);

			return producer;
		}catch (Exception e){
			log.error("启动生产者发生异常：{}", e);
		}finally {
			log.info("{}", logContext.toString());
		}
		return null;
	}

	public void testPushMessage(DefaultMQProducer producer, StringBuilder logContext){
		try {
			Message msg = new Message("test-topic" , "Tag", ("Hello RocketMQ").getBytes(RemotingHelper.DEFAULT_CHARSET));
			SendResult sendResult = producer.send(msg);
			LogSplicingUtils.addLogContext(logContext, "sendResult：%s", JsonUtils.toJsonString(sendResult));
		}catch (Exception e){
			log.error("{}", e);
		}
	}


}